with center_arrival_base as (--取t-2扫描时间 中心到件
    select waybill_no                  --运单编号
         , package_code                --所属包号
         , scan_site_code              --扫描网点code
         , scan_site as scan_site_name --扫描网点
         , network_type                --扫描网点类型，1.加盟商,2.转运中心,3.集散点,4.普通网点,5.代理区,6.总部,0.其他
         , scan_time                   --路段目的站点最晚发件时间
         , source                      --来源 PC端,自动分拣设备,移动端
         , site_agent_code
         , site_agent_name
    from (
             select waybill_no     --运单编号
                  , package_code   --所属包号
                  , scan_site_code --扫描网点code
                  , scan_site      --扫描网点
                  , network_type   --扫描网点类型，1.加盟商,2.转运中心,3.集散点,4.普通网点,5.代理区,6.总部,0.其他
                  , scan_time      --路段目的站点最晚发件时间
                  , source         --来源 PC端,自动分拣设备,移动端
                  , site_agent_code
                  , site_agent_name
                  , row_number() over (partition by waybill_no,scan_site_code order by scan_time asc) as rn
             from jms_dwd.dwd_tab_barscan_centerarrival_base_dt
             where dt >= date_add('{{ execution_date | cst_ds }}', -1) -- 数据为T-2的数据
               and dt <= '{{ execution_date | cst_ds }}'
               and waybill_no like 'JT%'
               and waybill_no is not null
               and waybill_no <> ''
               and network_type = 2                 --扫描网点类型，1.加盟商,2.转运中心,3.集散点,4.普通网点,5.代理区,6.总部,0.其他
               and date(scan_time) = date_add('{{ execution_date | cst_ds }}', -1)
         ) a
    where rn = 1
),
     center_send_base as (--取t-2扫描时间  中心发件 获取下一站
         select waybill_no        --运单编号
              , package_code      --所属包号
              , scan_site_code    --扫描网点code
              , scan_site         --扫描网点
              , scan_time         --路段目的站点最晚发件时间
              , next_station_code --下一站code
              , next_station      --下一站name
         from (
                  select waybill_no        --运单编号
                       , package_code      --所属包号
                       , scan_site_code    --扫描网点code
                       , scan_site         --扫描网点
                       , scan_time         --路段目的站点最晚发件时间
                       , next_station_code --下一站code
                       , next_station      --下一站name
                       , row_number() over (partition by waybill_no,scan_site_code order by scan_time desc) as rn
                  from jms_dwd.dwd_tab_barscan_centersend_base_dt
                  where dt >= date_add('{{ execution_date | cst_ds }}', -1)
                    and dt <= '{{ execution_date | cst_ds }}'
                    and scan_time >= date_add('{{ execution_date | cst_ds }}', -1)
                    and scan_time <= '{{ execution_date | cst_ds }}'
                    and waybill_no like 'JT%'
                    and network_type = 2 --扫描网点类型，1.加盟商,2.转运中心,3.集散点,4.普通网点,5.代理区,6.总部,0.其他
              ) a
         where rn = 1
     ),
     bagging_table as (--建包网点
         select *
         from (
                  select waybill_no --JT开头包号
                       , scan_site_code as                                                    pack_scan_site_code
                       , scan_site      as                                                    pack_scan_site_name
                       , scan_time
                       , row_number() over (partition by waybill_no,scan_site_code order by scan_time desc ) rnk
                  from jms_dwd.dwd_tab_barscan_bagging_base_dt
                  where dt >= date_add('{{ execution_date | cst_ds }}', -29)
                    and dt <= '{{ execution_date | cst_ds }}'
                    and scan_time <= date_add('{{ execution_date | cst_ds }}',-0)
                    and scan_time >= date_add('{{ execution_date | cst_ds }}',-1)
                    and waybill_no like 'JT%'
              )
         where rnk = 1
     ),
     arrival_send as (
         select t1.waybill_no                               --运单编号
              , t1.package_code                             --所属包号
              , t1.scan_site_code                           --扫描网点code
              , t1.scan_site_name                           --扫描网点
              , t1.network_type                             --扫描网点类型，1.加盟商,2.转运中心,3.集散点,4.普通网点,5.代理区,6.总部,0.其他
              , t1.scan_time                                --路段目的站点最晚发件时间
              , t1.source                                   --来源 PC端,自动分拣设备,移动端
              , t1.site_agent_code
              , t1.site_agent_name
              , t2.scan_time as      send_scan_time         --发件时间
              , t2.next_station_code send_next_station_code --下一站code
              , t2.next_station      send_next_station      --下一站name
         from center_arrival_base t1
                  left join center_send_base t2
                            on t1.waybill_no = t2.waybill_no
                                and t1.scan_site_code = t2.scan_site_code
                                and t1.scan_time < t2.scan_time
     )
insert
overwrite
table
jms_dm.dm_dev_inout_center_waybill_detail_dt
partition
(
dt
)
select t1.waybill_no                                                   as waybill_no              --运单编号
     , t1.package_code                                                 as package_code            --所属包号
     , t1.scan_site_code                                               as scan_site_code          --扫描网点code
     , t1.scan_site_name                                               as scan_site_name          --扫描网点
     , t1.site_agent_code                                              as site_agent_code         --代理区
     , t1.site_agent_name                                              as site_agent_name         --代理区
     , t1.network_type                                                 as network_type            --扫描网点类型，1.加盟商,2.转运中心,3.集散点,4.普通网点,5.代理区,6.总部,0.其他
     , t1.scan_time                                                    as scan_time               --路段目的站点最晚发件时间
     , t1.send_scan_time                                               as send_scan_time          --发件时间
     , t1.send_next_station_code                                       as send_next_station_code  --下一站code
     , t1.send_next_station                                            as send_next_station       --下一站name
     , t1.source                                                       as source                  --来源 PC端,自动分拣设备,移动端
     , t2.provider_id                                                  as arrival_provider_id     --到件中心省份id
     , t2.provider_desc                                                as arrival_provider_desc   --到件中心省份name
     , t3.provider_id                                                  as send_next_provider_id   --到件中心省份id
     , t3.provider_desc                                                as send_next_provider_desc --到件中心省份name
     , t4.pack_scan_site_code                                          as pack_scan_site_code     --建表网点
     , t4.pack_scan_site_name                                          as pack_scan_site_name     --建表网点
     , t6.package_code                                                 as send_package_code       --发件包号
     , if(t2.provider_id = t3.provider_id, 1, 0)                       as in_out_flag             --进出港标识,1代表进港 ,0代表出港(到件中心省份<>下一站的省份)
     , if(t1.package_code is not null and t1.package_code <> '', 1, 0) as is_pack_flag            --是否有包号,1有0没有
     , if(t1.scan_site_code = t4.pack_scan_site_code, 1, 0)            as is_build_pack_falg      --是否是建包中心,1是0否
     , if(t6.package_code is not null, 1, 0)                           as is_tear_pack_falg       --是否拆包1是0否
     , date_add('{{ execution_date | cst_ds }}', -1)                                      as date_time               --日期
     , date_add('{{ execution_date | cst_ds }}', -1)                                      as dt
from arrival_send t1
         left join jms_dim.dim_network_whole_massage t2
                   on t1.scan_site_code = t2.code
         left join jms_dim.dim_network_whole_massage t3
                   on t1.send_next_station_code = t3.code
         left join bagging_table t4
                   on t1.waybill_no = t4.waybill_no
                   and t1.scan_site_code = t4.pack_scan_site_code
         left join
     (
         select package_code
              , scan_site_code
              , scan_time
         from center_send_base
         where package_code is not null
         group by package_code
                , scan_site_code
                , scan_time
     ) t6
     on t1.package_code = t6.package_code
         and t1.scan_site_code = t6.scan_site_code
         and t1.scan_time < t6.scan_time
distribute by dt, abs(hash(waybill_no)) % 20
 ;
